package cn.kingyea.utils;

import java.io.File;
import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class HBaseCRUDTest {
	
	private final String ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME = "Client";
	private final String ZOOKEEPER_SERVER_PRINCIPAL_KEY = "zookeeper.server.principal";
	private final String ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL = "zookeeper/hadoop";

	private Configuration conf = null;
	private Connection conn = null;
	private TableName tableName = TableName.valueOf("hshe_test1");
	private String family = "info";
	
	private String krb5File = null;
	private String userName = null;
	private String userKeytabFile = null;
	
	/**
	 * 创建表
	 */
	@Test
	public void testCreateTable(){
		
		System.out.println("开始执行 testCreateTable()！");
		
		HColumnDescriptor hcd = new HColumnDescriptor(family);
		//设置数据块字符编码:DIFF,FAST_DIFF,PREFIX,PREFIX_TREE
		hcd.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);
		//设置数据压缩方式,HBase一共提供两种方式,GZ(高压缩率),SNAPPY(低压缩率)
		//高压缩率可能对数据检索性能有影响，一般使用SNAPPY
		hcd.setCompressionType(Compression.Algorithm.SNAPPY);
		
		HTableDescriptor htd = new HTableDescriptor(tableName);
		htd.addFamily(hcd);
		
		Admin admin = null;
		
		try {
			admin = conn.getAdmin();
			
			if(!admin.tableExists(tableName)){
				System.out.println("创建HTable:["+tableName.getNameAsString()+"]");
				admin.createTable(htd);
				System.out.println("创建HTable:["+tableName.getNameAsString()+"]成功！");
			}else{
				System.out.println("HTable:["+tableName.getNameAsString()+"]已存在！");
			}
			
		} catch (IOException e) {
			System.out.println("创建HTable:["+tableName.getNameAsString()+"]失败");
			e.printStackTrace();
		}finally {
			if(admin!=null){
				try {
					admin.close();
				} catch (IOException e) {
					System.out.println("关闭admin失败！");
					e.printStackTrace();
				}
			}
		}
		System.out.println("执行 testCreateTable() 完成！");
	}
	
	/**
	 * put方法
	 */
	@Test
	public void testPutData(){
		System.out.println("开始执行 testPutData()！");
		
		Table table = null;
		try {
			table = conn.getTable(tableName);
			
			List<Put> puts = DataGenerator.getPutList(Bytes.toBytes(family));
			table.put(puts);
			
		} catch (IOException e) {
			System.out.println("插入数据到HTable:["+tableName.getNameAsString()+"]失败！");
			e.printStackTrace();
		}finally {
			if(table!=null){
				try {
					table.close();
				} catch (IOException e) {
					System.out.println("关闭Table失败！");
					e.printStackTrace();
				}
			}
		}
		System.out.println("执行 testPutData() 完成！");
	}
	
	/**
	 * get方法
	 * @param rowKey 行键
	 * @param qualifiers 结果集
	 */
	@Test
	public void testGetData(){
		System.out.println("开始执行 testGetData()！");
		
		byte[] rowKey = Bytes.toBytes(String.format("%020d", 20));
		byte[][] qualifiers = {Bytes.toBytes("id"),Bytes.toBytes("name"),Bytes.toBytes("sex"),
				Bytes.toBytes("bdate"),Bytes.toBytes("id_code"),Bytes.toBytes("xzqh"),
				Bytes.toBytes("address"),Bytes.toBytes("created"),Bytes.toBytes("last_updated")};
		
		Table table = null;
		try {
			table = conn.getTable(tableName);
			Get get = new Get(rowKey);
			for(int i=0;i<qualifiers.length;i++){
				get.addColumn(Bytes.toBytes(family), qualifiers[i]);
			}
			
			Result result = table.get(get);
			System.out.println("列族:["+family+"],rowKey:["+Bytes.toString(rowKey)+"]");
			for(Cell cell: result.rawCells()){
				System.out.println("字段名:["+Bytes.toString(CellUtil.cloneQualifier(cell)) + "],字段值:[" + 
					Bytes.toString(CellUtil.cloneValue(cell))+"]");
			}
		} catch (IOException e) {
			System.out.println("查询失败！");
			e.printStackTrace();
		}finally {
			if(table!=null){
				try {
					table.close();
				} catch (IOException e) {
					System.out.println("关闭Table失败！");
					e.printStackTrace();
				}
			}
		}
		System.out.println("执行 testGetData() 完成！");
	}
	
	/**
	 * delete方法
	 */
	@Test
	public void testDelData(){
		System.out.println("开始执行 testDelData()！");
		byte[] rowKey = Bytes.toBytes(String.format("%020d", 20));
		Table table = null;
		try {
			table = conn.getTable(tableName);
			Delete del = new Delete(rowKey);
			table.delete(del);
			System.out.println("删除成功！");
		} catch (IOException e) {
			System.out.println("删除失败！");
			e.printStackTrace();
		}finally {
			if(table!=null){
				try {
					table.close();
				} catch (IOException e) {
					System.out.println("关闭Table失败！");
					e.printStackTrace();
				}
			}
		}
		System.out.println("执行 testDelData() 完成！");
	}
	
	/**
	 * 删除表
	 */
	@Test
	public void testDropTable(){
		System.out.println("开始执行 testDropTable()！");
		Admin admin = null;
		try {
			admin = conn.getAdmin();
			
			if(admin.tableExists(tableName)){
				System.out.println("删除HTable:["+tableName.getNameAsString()+"]");
				admin.disableTable(tableName);
				admin.deleteTable(tableName);
				System.out.println("删除HTable:["+tableName.getNameAsString()+"]成功！");
			}else{
				System.out.println("HTable:["+tableName.getNameAsString()+"]不存在！");
			}
			
		} catch (IOException e) {
			System.out.println("删除HTable:["+tableName.getNameAsString()+"]失败");
			e.printStackTrace();
		}finally {
			if(admin!=null){
				try {
					admin.close();
				} catch (IOException e) {
					System.out.println("关闭admin失败！");
					e.printStackTrace();
				}
			}
		}
		System.out.println("执行 testDropTable() 完成！");
	}
	
	@Test
	public void testGetTableDesc(){
		System.out.println("开始执行 testGetTableDesc()！");
		Admin admin = null;
		try {
			admin = conn.getAdmin();
			
			if(admin.tableExists(tableName)){
				HTableDescriptor tableDesc = admin.getTableDescriptor(tableName);
				System.out.println(tableDesc);
				HColumnDescriptor[] columnFamilies = tableDesc.getColumnFamilies();
				for(int i=0;i<columnFamilies.length;i++){
					System.out.println(columnFamilies[i].getNameAsString());
				}
				
				
			}else{
				System.out.println("HTable:["+tableName.getNameAsString()+"]不存在！");
			}
			
		} catch (IOException e) {
			System.out.println("删除HTable:["+tableName.getNameAsString()+"]失败");
			e.printStackTrace();
		}finally {
			if(admin!=null){
				try {
					admin.close();
				} catch (IOException e) {
					System.out.println("关闭admin失败！");
					e.printStackTrace();
				}
			}
		}
		System.out.println("执行 testGetTableDesc() 完成！");
	}
	
	@Before
	public void init(){
		try {
			System.setProperty("hadoop.home.dir", "E:\\SOFTWARE\\大数据");
			conf = new Configuration();
			String userdir = System.getProperty("user.dir") + File.separator;
//			userdir +=  "src" + File.separator + "test" + File.separator + "resources" + File.separator + "hadoop" + File.separator;
			userdir += "conf" + File.separator;
			System.out.println(userdir);
			conf.addResource(new Path(userdir + "core-site.xml"));
			conf.addResource(new Path(userdir + "hdfs-site.xml"));
			conf.addResource(new Path(userdir + "hbase-site.xml"));
			
			if (User.isHBaseSecurityEnabled(conf)) {
				userName = "dev_root";
				userKeytabFile = userdir + "user.keytab";
				krb5File = userdir + "krb5.conf";
				
				LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, userName, userKeytabFile);
				LoginUtil.setZookeeperServerPrincipal(ZOOKEEPER_SERVER_PRINCIPAL_KEY, ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL);
				LoginUtil.login(userName, userKeytabFile, krb5File, conf);
			}
			conn = ConnectionFactory.createConnection(conf);
			
		} catch (IOException e) {
			System.out.println("登录失败！");
			e.printStackTrace();
		}
		
	}
	@After
	public void close(){
		try {
			conn.close();
		} catch (IOException e) {
			System.out.println("关闭连接失败！");
			e.printStackTrace();
		}
	}
}
